Where we left off from the last Systematic Webinar¶
- Systematic Technical Analysis is easy with TA-Lib / Pandas TA¶
- Strategy backtesting is also relatively straightforward in Python¶
- Lets contrast say the output of 2 strategies¶
- SMA crossover¶
- 3Inside cdl strategy, only take signals in direction of SMA¶
SMA XOver¶
3Inside Cdl with SMA trend truncation¶
These types of result lead us to ask can we identify market regimes to help inform our strategy. We will look at a number of approaches to regime detection.
What is market regime?¶
'Clusters of persistent market conditions'
How many types of market regime are there?¶
calm, choppy, trending, crashing ....
How long do they last?¶
key question -
How do we know when they have changed?¶
key question - too early signal is noisy, too late signal is costly
How do we test the impact of regime on our strategy?¶
Approach 1 - What TA indicators can we bring to bear?¶
In [1]:
import refinitiv.data.eikon as ek
import talib
import pandas_ta as pta
import backtrader as bt
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.pyplot import figure
import matplotlib as mpl
import mplfinance as mpf
import datetime as dt
import numpy as np
from sklearn.svm import SVC
from statsmodels.tsa.stattools import adfuller
import warnings
warnings.simplefilter("ignore")
import configparser
cfg = configparser.ConfigParser()
cfg.read('rdp.cfg',encoding='utf-8')
%matplotlib inline
plt.style.use("dark_background")
mpl.style.use("dark_background")
ek.set_app_key(cfg['eikon']['app_key'])
#ek.set_app_key('cb7b5975d52943ff95e8d128fc87f5fcc88476f1')
In [126]:
df1= ek.get_timeseries('ESc1',fields=['OPEN','HIGH','LOW','CLOSE','VOLUME'],start_date='1997-01-01',
end_date='06-30-2022',interval='daily') #',
df1.columns=['open','high','low','close','volume'] #
df1
Out[126]:
In [127]:
df1.dropna(how="any", inplace=True)
len(df1)
Out[127]:
Pandas TA package is very useful tool to create various technical analysis features with ease¶
In [128]:
df1.ta.indicators()
ADX Indicator¶
In [129]:
adx = df1.copy()
adx.ta.adx(high=adx['high'], low=adx['low'], close=adx['close'], length=50, lensig=50, scalar=100, mamode="rma", drift=0, offset=0, append=True)
adx
Out[129]:
In [130]:
fig = plt.figure()
ax1 = fig.add_subplot(211)
ax2 = fig.add_subplot(212)
adx[['close']][1500:1700].plot(figsize=(18,15),ax=ax1)
adx[['ADX_50']][1500.plot(figsize=(18,15),ax=ax2) #'DMP_50','DMN_50'
Out[130]:
White's VHF Indicator¶
In [140]:
df2 = df1.copy()
df2.ta.vhf(close=df2['close'].astype(float), length=40, drift=0, offset=0, append=True)
df2
Out[140]:
In [150]:
fig = plt.figure()
ax1 = fig.add_subplot(211)
ax2 = fig.add_subplot(212)
df2[['close']][1500:1700].plot(figsize=(18,15),ax=ax1)
df2[['VHF_40']][1500:1700].plot(figsize=(18,15),ax=ax2) #'DMP_50','DMN_50'
Out[150]:
In [147]:
df3 = df1.copy()
#close=None, length=None, bars=None, offset=None,
df3.ta.ebsw(close=df3['close'].astype(float), length=40, bars=10, offset=0, append=True)
df3
Out[147]:
In [149]:
fig = plt.figure()
ax1 = fig.add_subplot(211)
ax2 = fig.add_subplot(212)
df3[['close']][1500:1700].plot(figsize=(18,15),ax=ax1)
df3[['EBSW_40_10']][1500:1700].plot(figsize=(18,15),ax=ax2) #'DMP_50','DMN_50'
Out[149]:
In [ ]:
In [ ]:
In [144]:
df4 = df1.copy()
df4.ta.chop(close=df4['close'].astype(np.int64), length=50, atr_length =50, scalar=0, drift=0, offset=0, append=True)
df4
Out[144]:
In [146]:
fig = plt.figure()
ax1 = fig.add_subplot(211)
ax2 = fig.add_subplot(212)
df4[['close']].plot(figsize=(18,15),ax=ax1)
df4[['CHOP_50_50_100']].plot(figsize=(18,15),ax=ax2)
Out[146]:
In [3]:
df5 = df1.copy()
In [13]:
import ruptures as rpt
points = np.array(df5['close'])
In [14]:
#Changepoint detection with the Pelt search method
#model="rbf"
algo = rpt.Pelt(model="rbf").fit(points)
result = algo.predict(pen=10)
rpt.display(points, result, figsize=(8, 4))
plt.title('Change Point Detection: Pelt Search Method')
plt.show()
In [ ]:
#Changepoint detection with the Binary Segmentation search method
#model = "l2"
algo = rpt.Binseg(model="l2").fit(points)
my_bkps = algo.predict(n_bkps=10)
# show results
rpt.show.display(points, my_bkps, figsize=(8, 4))
plt.title('Change Point Detection: Binary Segmentation Search Method')
plt.show()
In [ ]:
#Changepoint detection with window-based search method
#model = "l2"
algo = rpt.Window(width=40, model="l2").fit(points)
my_bkps = algo.predict(n_bkps=10)
rpt.show.display(points, my_bkps, figsize=(8, 4))
plt.title('Change Point Detection: Window-Based Search Method')
plt.show()
In [ ]:
#Changepoint detection with dynamic programming search method
#model = "l1"
algo = rpt.Dynp(model="l1", min_size=3, jump=5).fit(points)
my_bkps = algo.predict(n_bkps=10)
rpt.show.display(points, my_bkps, figsize=(8, 4))
plt.title('Change Point Detection: Dynamic Programming Search Method')
plt.show()
Online CPD using changefinder - Haykaz¶
In [4]:
points = df4['close']
In [20]:
%run ./regime_detection.ipynb
In [120]:
import changefinder
#import plotly.express as px
cf = changefinder.ChangeFinder()
scores = [cf.update(p) for p in points]
fig = plt.figure()
ax1 = fig.add_subplot(211)
ax2 = fig.add_subplot(212)
df4[['close']].plot(figsize=(18,15),ax=ax1)
pd.DataFrame(scores).plot(figsize=(18,15),ax=ax2)
Out[120]:
In [118]:
mean = np.mean(scores)
std = np.std(scores)
states = []
for score in scores:
#no vol
if score > (mean - 2*std) and score < (mean + 2*std):
states.append(0)
#low vol
# elif score > (mean - 2*std) and score < (mean + 2*std):
# states.append(1)
#high vol
else:
states.append(1)
In [119]:
plot_hidden_states_cf(np.array(states), df4[['close']],2)
In [ ]:
Approach 3 - Unsupervised Learning Approaches - K-means, Agglomerative, HMM, GMM et al¶
To capture the dynamic behavior of such time-series, it is often preferable to employ multiple models. Particularly, when attempting to model financial time-series over a set of regimes, the Markov switching model [18], also known as the regime switching model, has been studied and applied. Generally, the Markov switching model is governed by an unobservable state variable and time-varying transition probabilities. Despite the many empirical applications and theoretical developments involving Markov switching models, a necessary condition for the method is the need to know the number of regimes which exist prior to estimating model coefficients. This limits the flexibility of the regime switching model in two contexts. The first being applications where modeling and forecasting the time-series may be undesirable, and identifying the occurrence of the underlying regimes may be all that is sought after. The second consists of contexts where it may be beneficial to directly infer the number of regimes from data.
Data Ingestion and Engineering¶
In [53]:
trading_instrument = 'ESc1'
instruments = [trading_instrument]
In [54]:
prices = rd.get_history(
universe = instruments,
fields=['TRDPRC_1'],
#fields=['ASIA_CLOSE'],
start = "1997-01-01",
end = "2022-12-05",
).dropna()
prices.sort_values(by = 'Date', inplace = True)
In [55]:
prices.columns.name = trading_instrument
prices = prices.iloc[:, :-1]
In [56]:
prices.iloc[1782] = (prices.iloc[1781] + prices.iloc[1783])/2
prices
Out[56]:
In [57]:
prices_change, prices, split_index = DataEngineering(prices, split_date = '2006-01-01').prepare_data()
In [58]:
prices_change
Out[58]:
Regime Detection - Modeling and In Sample testing¶
In [59]:
regime_detection = RegimeDetection()
In [17]:
clustering = regime_detection.get_regimes_clustering('AgglomerativeClustering')
clustering_states = clustering.fit_predict(prices)
plot_hidden_states(np.array(clustering_states), prices_change[[f'{trading_instrument}_close']],2)
In [18]:
clustering = regime_detection.get_regimes_clustering('kmeans')
clustering_states = clustering.fit_predict(prices)
plot_hidden_states(np.array(clustering_states), prices_change[[f'{trading_instrument}_close']],2)
In [22]:
gmm_model = regime_detection.get_regimes_gmm(prices)
gmm_states = gmm_model.predict(prices)
plot_hidden_states(np.array(gmm_states), prices_change[[f'{trading_instrument}_close']],2)
In [23]:
hmm_model = regime_detection.get_regimes_hmm(prices)
hmm_states = hmm_model.predict(prices)
plot_hidden_states(np.array(hmm_states), prices_change[[f'{trading_instrument}_close']],2)
Regime detection - Out of Sample testing¶
In [24]:
states_pred_gmm = feed_forward_training('gmm', prices, split_index, 20)
In [25]:
plot_hidden_states(np.array(states_pred_gmm), prices_change[[f'{trading_instrument}_close']][split_index:],2)
In [60]:
states_pred_hmm = feed_forward_training('hmm', prices, split_index, 20)
In [27]:
plot_hidden_states(np.array(states_pred_hmm), prices_change[[f'{trading_instrument}_close']][split_index:],2)
In [ ]:
Strategy Implementation¶
In [61]:
prices_open = rd.get_history(
universe = 'ESc1',
fields=[ 'OPEN_PRC'],
start = "1997-09-09",
end = "2022-11-24",).dropna()
prices_open.sort_values(by = 'Date', inplace = True)
prices_open = prices_open.iloc[:, :-1]
prices_open.columns.name = trading_instrument
In [62]:
prices_open = prices_open[split_index:]
prices_open = prices_open.shift(-1)
prices_open.head()
Out[62]:
In [63]:
prices = pd.DataFrame(prices_change[split_index:][f'{trading_instrument}_close'])
prices['State'] = states_pred_hmm
In [64]:
prices = pd.DataFrame(prices['State']).merge(prices_open, on = 'Date')
prices.reset_index(inplace=True)
prices.dropna(inplace=True)
prices.columns.name = trading_instrument
prices
Out[64]:
HMM Strategy¶
In [65]:
strategy_outcome = HMMStrategy(prices, 2).run()
strategy_outcome
Out[65]:
In [66]:
strategy_outcome['Current Price'] = strategy_outcome['Current Price'] - 1278.25
StrategyOutcomePlot(strategy_outcome).plot(outcome = 'P&L', signal_dot = 'Current Price')
MA Strategy with HMM¶
In [34]:
strategy_outcome = MovingaAveragesStrategy(prices, 10, 30, True, 2).run()
strategy_outcome
Out[34]:
In [35]:
strategy_outcome['Current Price'] = strategy_outcome['Current Price'] - 1278.25
StrategyOutcomePlot(strategy_outcome).plot(outcome = 'P&L', signal_dot = 'Current Price')
In [67]:
strategy_outcome_hmm = HMMStrategy(prices,3).run()
strategy_outcome_ma = MovingaAveragesStrategy(prices, 10, 30, False, 2).run()
strategy_outcome_ma_hmm = MovingaAveragesStrategy(prices, 10, 30, True, 2).run()
In [68]:
strategies = strategy_outcome_hmm[['Current Price', 'Portfolio Valuation', 'P&L']].merge(
strategy_outcome_ma[['Current Price','Portfolio Valuation', 'P&L']].merge(
strategy_outcome_ma_hmm[['Current Price','Portfolio Valuation', 'P&L']], on = 'Date', suffixes = ['_ma', '_ma_hmm']), on = 'Date')
strategies['Buy and Hold'] = strategies['Current Price'] - strategies['Current Price'][0]
strategies
Out[68]:
In [70]:
fig = make_subplots(specs=[[{"secondary_y": True}]])
fig.add_trace(go.Line(x=strategies.index, y=strategies["Buy and Hold"], name = 'Buy and Hold',
line_color = 'black'),secondary_y=True)
fig.add_trace(go.Line(x= strategy_outcome.index, y=strategies['P&L_ma_hmm'], name = 'P&L for MA with hmm',
line_color = 'khaki'), secondary_y=False)
fig.add_trace(go.Line(x= strategy_outcome.index, y=strategies['P&L_ma'], name = 'P&L for MA without hmm',
line_color = 'grey'), secondary_y=False)
fig.add_trace(go.Line(x= strategy_outcome.index, y=strategies['P&L'], name = 'P&L HMM',
line_color = 'green'), secondary_y=False)
fig.add_trace(go.Line(x= strategy_outcome.index, y=strategies['Portfolio Valuation_ma_hmm'], name = 'Valuation for MA with hmm',
line_color = 'cyan'), secondary_y=True)
fig.add_trace(go.Line(x= strategy_outcome.index, y=strategies['Portfolio Valuation_ma'], name = 'Valuation for MA without hmm',
line_color = 'blue'), secondary_y=True)
fig.add_trace(go.Line(x= strategy_outcome.index, y=strategies['Portfolio Valuation'], name = 'Valuation HMM',
line_color = 'brown'), secondary_y=True)
fig.update_layout(height=400, width=900, legend=dict(
yanchor="top", y=0.99,
xanchor="left",x=0.01),
margin=dict(l=20, r=20, t=20, b=20))
fig.show()
In [ ]: